package com.tarena.rocketmq.sequence;

import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

@Slf4j
public class SequenceProducer {
    public static void main(String[] args) {
        //1.创建rocketmq默认生产者
        DefaultMQProducer producer = new DefaultMQProducer();
        //2.设置localhost:9876 设置分组normal-group-prod
        producer.setNamesrvAddr("localhost:9876");
        producer.setProducerGroup("sequence-group-prod");
        //3.准备发送消息message 包含字符串body
        String body = "hello rocketmq";
        //4.启动生产者并且发送消息
        SendResult result = null;
        //5.构建一个消息选择器
        MessageQueueSelector selector = new MessageQueueSelector() {
            /**
             * @param mqs 当前主题可用队列 默认4个
             * @param msg 当前消息
             * @param arg 自定义选择器计算的业务数据 比如订单编号
             * @return
             */
            @Override public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                if (arg instanceof String) {
                    String orderId = (String) arg;
                    //定义一个hash取余计算
                    if (CollectionUtils.isNotEmpty(mqs)){
                        return mqs.get(hash(orderId,mqs.size()));
                    }
                }
                return mqs.get(0);
            }
        };
        try {
            //6.启动生产者
            producer.start();
            //7.发送消息
            for (int i = 0; i < 10; i++) {
                String msg=body+" "+i;
                Message message = new Message("sequence-topic",msg.getBytes());

                //8.发送消息
                result = producer.send(message,selector,"同一个业务id");
                log.info("发送消息结果:{}",result);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
    public static Integer hash(String businessId,Integer mod){
        Integer index =(businessId.hashCode()&Integer.MAX_VALUE)%mod;
        log.info("当前下标:{}",index);
        return index;
    }
}
